﻿using System;
using System.Collections.Generic;
using System.Text;
using System.Net;
using System.Net.Sockets;
using System.IO;

namespace Uniriotec.DC.CoordDist.Distribuida
{

    public class Process
    {
        #region Private Variables

        private bool requesting = false;
        private int id = 0;
        private int port = 0;
        private int time = 0;
        private int higherTime = 0;
        private List<int> nodeIDs = new List<int>();
        private Dictionary<int, bool> deferredReplys = new Dictionary<int, bool>();

        private int ackCount = 0;
        private bool onCritical = false;
        private Socket sockListener;
        private Socket sockWorker;
        private int totalProcs = 0;

        #endregion

        #region Initialization

        private void Init()
        {
            sockListener = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            IPEndPoint local = new IPEndPoint(IPHelper.GetIP("localhost"), port);
            sockListener.Bind(local);
            sockListener.Listen(200);
            sockListener.BeginAccept(new AsyncCallback(OnClientConnect), sockListener);

            for (int i = 1; i <= totalProcs; i++)
                nodeIDs.Add(i);

            for (int i = 1; i <= totalProcs; i++)
                deferredReplys.Add(i, false);
        }

        public void Start(object startParameters)
        {
            StartProcessParameters parameters = (StartProcessParameters)startParameters;
            this.id = parameters.Id;
            this.higherTime = 1;//= parameters.Id;
            this.port = parameters.Port;
            this.totalProcs = parameters.TotalProcs;

            Init();
            DoWork();
        }

        #endregion

        #region Helper Methods

        private void Debug(string str)
        {
            Console.WriteLine("(Process " + id + ") " + str);
        }

        private int GetPort(int process)
        {
            return port - (id - process);
        }


        #endregion

        #region Connection

        public void OnClientConnect(IAsyncResult asyn)
        {
            try
            {
                sockWorker = sockListener.EndAccept(asyn);
                StateObject state = new StateObject();
                state.workSocket = sockWorker;

                sockWorker.BeginReceive(state.buffer, 0, StateObject.BufferSize, SocketFlags.None, new AsyncCallback(OnDataReceived), state);

            }
            catch (Exception e)
            {
                throw e;
            }
        }

        #endregion

        #region Receive Data

        public void OnDataReceived(IAsyncResult asyn)
        {
            StateObject so = (StateObject)asyn.AsyncState;
            int read = sockWorker.EndReceive(asyn);

            if (read > 0)
            {
                so.sb.Append(Encoding.ASCII.GetString(so.buffer, 0, read));
                so.workSocket.BeginReceive(so.buffer, 0, StateObject.BufferSize, 0,
                                         new AsyncCallback(OnDataReceived), so);
            }
            else
            {
                if (so.sb.Length > 1)
                {
                    //All of the data has been read
                    string strContent;
                    strContent = so.sb.ToString();

                    string[] partitions = strContent.Split(',');
                    if (partitions.Length > 3)
                        throw new Exception("Wrong Message (proc " + id + "): " + strContent);

                    Message msg = new Message(Convert.ToInt32(partitions[0]), Convert.ToInt32(partitions[1]), (MessageVerb)Convert.ToInt32(partitions[2]));
                    ReceiveMessage(msg);

                }
                so.workSocket.Close();
                sockListener.BeginAccept(new AsyncCallback(OnClientConnect), sockListener);
            }

        }

        private void ReceiveMessage(Message message)
        {
            switch (message.Verb)
            {
                case MessageVerb.Ack: ackCount++;
                    Debug("received " + ackCount + " ACKs - Last ACK from " + message.ProcId);
                    break;
                case MessageVerb.Req: ReceiveReq(message);
                    Debug("received Req from (Process: " + message.ProcId + ") Time: " + message.Time);
                    break;
            }
        }

        private void ReceiveReq(Message message)
        {
            bool defer = IsDeferred(message);

            if (message.Time > higherTime)
                higherTime = message.Time;

            if (defer)
            {
                deferredReplys[message.ProcId] = true;
                return;
            }

            SendMessage(MessageVerb.Ack, id, message.ProcId, time);
        }

        #endregion

        #region Send Data

        private void SendToAll(MessageVerb verb, int from)
        {
            Debug("sent to all: " + verb.ToString());

            foreach (int nodeId in nodeIDs)
            {
                if (nodeId == id)
                    continue;
                SendMessage(verb, id, nodeId, time);
            }
        }

        private void SendMessage(MessageVerb verb, int from, int to, int time)
        {

            Message message = new Message(from, time, verb);

            Debug("TO: " + to + " sent : " + message.ToString());
            SendMessage(message, to);
        }

        private void SendMessage(Message message, int to)
        {
            int remotePort = GetPort(to);

            Socket senderSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
            senderSocket.Connect("localhost", remotePort);
            Send(senderSocket, message.ToString());
        }

        private static void Send(Socket handler, String data)
        {
            // Convert the string data to byte data using ASCII encoding.
            byte[] byteData = Encoding.ASCII.GetBytes(data);

            // Begin sending the data to the remote device.
            handler.BeginSend(byteData, 0, byteData.Length, 0,
            new AsyncCallback(SendCallback), handler);
        }

        private static void SendCallback(IAsyncResult ar)
        {
            try
            {
                // Retrieve the socket from the state object.
                Socket handler = (Socket)ar.AsyncState;

                // Complete sending the data to the remote device.
                int bytesSent = handler.EndSend(ar);

                handler.Shutdown(SocketShutdown.Both);
                handler.Close();
            }
            catch (Exception e)
            {
                throw e;
            }
        }

        #endregion

        #region Misc Algorithm Methods

        private void ChooseTime()
        {
            requesting = true;
            time = higherTime + 1;
        }

        private void AskToEnterCritical()
        {
            Debug("asked to enter critical");
            ChooseTime();
            ackCount = 0;
            SendToAll(MessageVerb.Req, id);
        }

        private bool IsDeferred(Message message)
        {
            if (onCritical)
                return true;
            if (requesting && (time < message.Time || (time == message.Time && id < message.ProcId)))
                return true;
            return false;
        }

        private void WaitToEnterCritical()
        {
            Debug("is waiting to enter critical");

            while (ackCount != nodeIDs.Count - 1)
            {
                System.Threading.Thread.Sleep(500);
            }
        }

        private void ReleaseLock()
        {
            onCritical = false;
            Debug("released lock");
            ReplyDeferred();
        }

        private void ReplyDeferred()
        {
            requesting = false;
            Message reply = new Message(id, time, MessageVerb.Ack);

            for (int i = 1; i <= totalProcs; i++)
            {
                if (deferredReplys[i])
                {
                    deferredReplys[i] = false;
                    SendMessage(reply, i);
                }

            }
        }

        #endregion

        #region Work Methods

        private void DoNormalWork()
        {
            Random rnd = new Random();
            System.Threading.Thread.Sleep(rnd.Next(20, 4000));
        }

        private void DoWork()
        {
            DoNormalWork();

            int j = 0;
            while (j++ < 10)
            {
                DoNormalWork();
                AskToEnterCritical();
                WaitToEnterCritical();
                DoCriticalWork();
            }

            Debug("finished doing work");
        }

        private void DoCriticalWork()
        {
            onCritical = true;

            Debug("is doing critical work");

            FileStream file = new FileStream(@"C:\distributed.txt", FileMode.Append, FileAccess.Write);
            StreamWriter writer = new StreamWriter(file);
            writer.WriteLine("Process " + id + " did critical work");

            System.Threading.Thread.Sleep(1500);

            writer.Flush();
            file.Close();


            ReleaseLock();
        }

        #endregion

    }

    #region StartProcessParameters class

    public class StartProcessParameters
    {

        private int id = 0;
        private int port = 0;
        private int totalProcs = 0;

        public int TotalProcs
        {
            get { return totalProcs; }
            set { totalProcs = value; }
        }

        public int Port
        {
            get { return port; }
            set { port = value; }
        }

        public int Id
        {
            get { return id; }
            set { id = value; }
        }

        public StartProcessParameters(int totalProcs, int id, int port)
        {
            this.totalProcs = totalProcs;
            this.id = id;
            this.port = port;
        }

    }

    #endregion

}
